package com.apes.pi.parser;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apes.framework.api.SpringManager;
import com.apes.framework.config.redis.RedisMessageListener;
import com.apes.framework.plugin.cache.Cache;
import com.apes.framework.plugin.esb.api.EsbRouter;
import com.apes.framework.util.ApesConst;
import com.apes.framework.util.EventUtil;
import com.apes.framework.util.MapUtil;
import com.apes.pi.model.*;
import com.apes.pi.repository.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 功能：
 *
 * @author xul
 * @create 2019-11-08 16:23
 */
@Component
public class QueueCacheManager {

    private final String CONST_QUEUE_CACHE = "piQueueCache";
    private final String CONST_QUEUE_KEY = "Queue";
    private final String CONST_QUEUE_RECEIVER_KEY = "QueueReceiver";
    private final String CONST_PIPE_KEY = "Pipe";
    private final String CONST_RECEIVER_KEY = "Receiver";
    private final String CONST_CONNECTOR_KEY = "Connector";

    @Autowired
    private Cache cache;

    @Autowired
    private PiQueueRepository piQueueRepository;

    @Autowired
    private QueueReceiverRepository queueReceiverRepository;

    @Autowired
    private PipeRepository pipeRepository;

    @Autowired
    private StringRedisTemplate template;

    @RedisMessageListener(topic = "pi.channel")
    public void subscribe(String message) {
        JSONObject jo = JSON.parseObject(message);
        switch (jo.getString("topic")) {
            case "addQueue":
                PiQueue queue = getQueues().get(jo.get("queueId"));
                EsbRouter.setRouter(queue.methodName(), ApesConst.METHOD_VERSION, new PiQueueElement(queue));
                break;
            case "removeQueue":
                EsbRouter.removeRouter(jo.getString("method"), ApesConst.METHOD_VERSION);
                break;
            case "addQueueReceiver":
                QueueReceiver queueReceiver = getQueueReceivers().get(jo.get("queueReceiverId"));
                SpringManager.getBean(PiEventListener.class).subscribe(queueReceiver);
                break;
            case "removeQueueReceiver":
                EventUtil.sendAynEvent(this, "event:mq.unsubscribe", MapUtil.mapper("topic", jo.get("mq.topic")));
                break;
            default:
                break;
        }
    }

    public void addQueue(PiQueue queue) {
        refreshQueue(queue, false);
        Map<String, PiQueue> queues = getQueues();
        queues.put(queue.getId(), queue);
        cache.put(CONST_QUEUE_CACHE, CONST_QUEUE_KEY, queues);
        template.convertAndSend("pi.channel", JSON.toJSONString(MapUtil.mapper("topic", "addQueue", "queueId", queue.getId())));
    }

    private void refreshQueue(PiQueue queue, boolean removed) {
        getPipeByParentId(queue.getId(), CONST_QUEUE_KEY).forEach(pipe -> removePipe(pipe.getId()));
        if (!removed) {
            if (queue.getBeforePipes() != null) queue.getBeforePipes().forEach(this::addPipe);
            if (queue.getAfterPipes() != null)  queue.getAfterPipes().forEach(this::addPipe);
        }
    }

    public void removeQueue(String queueId) {
        Map<String, PiQueue> queues = getQueues();
        PiQueue queue = queues.get(queueId);
        refreshQueue(queue, true);
        if (queue == null) return;
        queues.remove(queueId);
        cache.put(CONST_QUEUE_CACHE, CONST_QUEUE_KEY, queues);
        template.convertAndSend("pi.channel", JSON.toJSONString(MapUtil.mapper("topic", "removeQueue", "queueId", queueId, "method", queue.methodName())));
    }

    public Map<String, PiQueue> getQueues() {
        String CACHE_INIT_FLAG = "QueueCacheInitFlag";
        Boolean init = (Boolean) cache.get(CONST_QUEUE_CACHE, CACHE_INIT_FLAG);
        if (init == null || !init) {
            execute(() -> {
                Map<String, PiQueue> queues = piQueueRepository.findAll().stream().collect(Collectors.toMap(PiQueue::getId, piQueue -> piQueue));
                cache.put(CONST_QUEUE_CACHE, CONST_QUEUE_KEY, queues);
                cache.put(CONST_QUEUE_CACHE, CACHE_INIT_FLAG, true);
            });
        }
        return (Map<String, PiQueue>) cache.get(CONST_QUEUE_CACHE, CONST_QUEUE_KEY);
    }

    public void addQueueReceiver(QueueReceiver queueReceiver) {
        refreshQueueReceiver(queueReceiver, false);
        Map<String, QueueReceiver> queueReceivers = getQueueReceivers();
        queueReceivers.put(queueReceiver.getId(), queueReceiver);
        cache.put(CONST_QUEUE_CACHE, CONST_QUEUE_RECEIVER_KEY, queueReceivers);
        template.convertAndSend("pi.channel", JSON.toJSONString(MapUtil.mapper("topic", "addQueueReceiver", "queueReceiverId", queueReceiver.getId())));
    }

    private void refreshQueueReceiver(QueueReceiver queueReceiver, boolean removed) {
        getPipeByParentId(queueReceiver.getId(), CONST_QUEUE_RECEIVER_KEY).forEach(
                pipe -> removePipe(pipe.getId())
        );
        if (!removed) {
            if (queueReceiver.getBeforePipes()!=null) queueReceiver.getBeforePipes().forEach(this::addPipe);
            if (queueReceiver.getAfterPipes()!=null) queueReceiver.getAfterPipes().forEach(this::addPipe);
        }
    }

    public void removeQueueReceiver(String queueReceiverId) {
        Map<String, QueueReceiver> queueReceivers = getQueueReceivers();
        QueueReceiver queueReceiver = queueReceivers.get(queueReceiverId);
        refreshQueueReceiver(queueReceiver, true);
        if (queueReceiver == null) return;
        queueReceivers.remove(queueReceiverId);
        cache.put(CONST_QUEUE_CACHE, CONST_QUEUE_RECEIVER_KEY, queueReceivers);
        template.convertAndSend("pi.channel", JSON.toJSONString(MapUtil.mapper("topic", "removeQueueReceiver", "queueReceiverId", queueReceiverId, "mq.topic", queueReceiver.topicName())));
    }

    public Map<String, QueueReceiver> getQueueReceivers() {
        String CACHE_INIT_FLAG = "QueueReceiverCacheInitFlag";
        Boolean init = (Boolean) cache.get(CONST_QUEUE_CACHE, CACHE_INIT_FLAG);
        if (init == null || !init) {
            execute(() -> {
                Map<String, QueueReceiver> queueReceivers = queueReceiverRepository.findAll().stream()
                        .collect(Collectors.toMap(QueueReceiver::getId, queueReceiver -> queueReceiver));
                cache.put(CONST_QUEUE_CACHE, CONST_QUEUE_RECEIVER_KEY, queueReceivers);
                cache.put(CONST_QUEUE_CACHE, CACHE_INIT_FLAG, true);
            });
        }
        return (Map<String, QueueReceiver>) cache.get(CONST_QUEUE_CACHE, CONST_QUEUE_RECEIVER_KEY);
    }

    public List<QueueReceiver> getQueueReceiversByQueueId(String queueId) {
        List<QueueReceiver> queueReceivers = new ArrayList<>();
        getQueueReceivers().forEach((k, queueReceiver) -> {
            if (queueReceiver.getQueue().getId().equals(queueId)) queueReceivers.add(queueReceiver);
        });
        queueReceivers.sort(Comparator.comparing(QueueReceiver::getId));
        return queueReceivers;
    }

    public void addPipe(Pipe pipe) {
        Map<String, Pipe> pipes = getPipes();
        pipes.put(pipe.getId(), pipe);
        cache.put(CONST_QUEUE_CACHE, CONST_PIPE_KEY, pipes);
    }

    public void removePipe(String pipeId) {
        Map<String, Pipe> pipes = getPipes();
        pipes.remove(pipeId);
        cache.put(CONST_QUEUE_CACHE, CONST_PIPE_KEY, pipes);
    }

    public Map<String, Pipe> getPipes() {
        String CACHE_INIT_FLAG = "PipeCacheInitFlag";
        Boolean init = (Boolean) cache.get(CONST_QUEUE_CACHE, CACHE_INIT_FLAG);
        if (init == null || !init) {
            execute(() -> {
                Map<String, Pipe> queues = pipeRepository.findAll().stream().collect(Collectors.toMap(Pipe::getId, pipe -> pipe));
                cache.put(CONST_QUEUE_CACHE, CONST_PIPE_KEY, queues);
                cache.put(CONST_QUEUE_CACHE, CACHE_INIT_FLAG, true);
            });
        }
        Map<String, Pipe> m = (Map<String, Pipe>) cache.get(CONST_QUEUE_CACHE, CONST_PIPE_KEY);
        if (m == null) return new HashMap();
        return m;
    }

    private List<Pipe> getPipeByParentId(String parentId, String parentType) {
        List<Pipe> pipes = new ArrayList<>();
        getPipes().forEach((k, pipe) -> {
            if (parentId.equals(pipe.getParent()) && parentType.equals(pipe.getParentType())) pipes.add(pipe);
        });
        pipes.sort(Comparator.comparing(Pipe::getId));
        return pipes;
    }

    public List<Pipe> getPipeByParentId(String parentId, String position, String parentType) {
        List<Pipe> pipes = new ArrayList<>();
        getPipes().forEach((k, pipe) -> {
            if (pipe.getParent().equals(parentId) && pipe.getPosition().equals(position)
                    && pipe.getParentType().equals(parentType)) pipes.add(pipe);
        });
        pipes.sort(Comparator.comparing(Pipe::getId));
        return pipes;
    }

    public void addReceiver(Receiver receiver) {
        refreshReceiver(receiver, false);
        Map<String, Receiver> queueReceivers = getReceivers();
        queueReceivers.put(receiver.getId(), receiver);
        cache.put(CONST_QUEUE_CACHE, CONST_RECEIVER_KEY, queueReceivers);
    }

    private void refreshReceiver(Receiver receiver, boolean removed) {
        getPipeByParentId(receiver.getId(), CONST_RECEIVER_KEY).forEach(pipe -> removePipe(pipe.getId()));
        if (!removed) {
            receiver.getParams().size();
            receiver.getConnector().getId();
            if (receiver.getBeforePipes() != null) receiver.getBeforePipes().forEach(this::addPipe);
            if (receiver.getAfterPipes() != null) receiver.getAfterPipes().forEach(this::addPipe);
        }
    }

    public void removeReceiver(String receiverId) {
        Map<String, Receiver> receivers = getReceivers();
        Receiver receiver = receivers.get(receiverId);
        refreshReceiver(receiver, true);
        if (receiver == null) return;
        receivers.remove(receiverId);
        cache.put(CONST_QUEUE_CACHE, CONST_RECEIVER_KEY, receivers);
    }

    public Map<String, Receiver> getReceivers() {
        String CACHE_INIT_FLAG = "ReceiverCacheInitFlag";
        Boolean init = (Boolean) cache.get(CONST_QUEUE_CACHE, CACHE_INIT_FLAG);
        if (init == null || !init) {
            execute(() -> {
                Map<String, Receiver> receivers = SpringManager.getBean(ReceiverRepository.class).findAll().stream()
                        .collect(Collectors.toMap(Receiver::getId, receiver -> {
                            refreshReceiver(receiver, false);
                            return receiver;
                        }));
                cache.put(CONST_QUEUE_CACHE, CONST_RECEIVER_KEY, receivers);
                cache.put(CONST_QUEUE_CACHE, CACHE_INIT_FLAG, true);
            });
        }
        return (Map<String, Receiver>) cache.get(CONST_QUEUE_CACHE, CONST_RECEIVER_KEY);
    }


    public void addPiConnector(PiConnector connector) {
        refreshPiConnector(connector, false);
        Map<String, PiConnector> connectors = getPiConnectors();
        connectors.put(connector.getId(), connector);
        cache.put(CONST_QUEUE_CACHE, CONST_CONNECTOR_KEY, connectors);
    }

    private void refreshPiConnector(PiConnector connector, boolean removed) {
        if (!removed) {
            connector.getProtocol().getCode();
            connector.getParams().stream().forEach(params -> {
                params.getCategory().getCode();
                params.getType().getCode();
            });
        }
    }

    public void removePiConnector(String connectorId) {
        Map<String, PiConnector> connectors = getPiConnectors();
        PiConnector connector = connectors.get(connectorId);
        refreshPiConnector(connector, true);
        if (connector == null) return;
        connectors.remove(connectorId);
        cache.put(CONST_QUEUE_CACHE, CONST_CONNECTOR_KEY, connectors);
    }

    public Map<String, PiConnector> getPiConnectors() {
        String CACHE_INIT_FLAG = "ConnectorCacheInitFlag";
        Boolean init = (Boolean) cache.get(CONST_QUEUE_CACHE, CACHE_INIT_FLAG);
        if (init == null || !init) {
            execute(() -> {
                Map<String, PiConnector> connectors = SpringManager.getBean(PiConnectorRepository.class).findAll().stream()
                        .collect(Collectors.toMap(PiConnector::getId, connector -> {
                            refreshPiConnector(connector, false);
                            return connector;
                        }));
                cache.put(CONST_QUEUE_CACHE, CONST_CONNECTOR_KEY, connectors);
                cache.put(CONST_QUEUE_CACHE, CACHE_INIT_FLAG, true);
            });
        }
        return (Map<String, PiConnector>) cache.get(CONST_QUEUE_CACHE, CONST_CONNECTOR_KEY);
    }


    @Autowired
    private TransactionTemplate transactionTemplate;

    public void execute(Callback callback) {
        transactionTemplate.execute(status -> {
            try {
                callback.execute();
                return null;
            } catch (Exception e) {
                status.isRollbackOnly();
                throw e;
            }
        });
    }

    @FunctionalInterface
    public interface Callback {
        void execute();
    }
}
